# -*- coding: utf-8 -*-
import pendulum
from datetime import timedelta, datetime
from airflow import DAG
from airflow.utils.trigger_rule import TriggerRule
from airflow.models.variable import Variable
from utils import combine
from utils.alerts import dingding_dag_alert, dingding_task_alert, dingding_sla_alert, call_phone_by_email, call_phone_on_sla_miss
from utils.macros import macros_dict
from utils.alerts import feishu_dag_alert,feishu_task_alert,feishu_sla_alert
import jinja2

is_pro_env = Variable.get(key='env', default_var='').upper() == 'PRO'

with DAG(dag_id='jms_data_back_up',
         start_date=datetime(2021, 3, 11, tzinfo=pendulum.timezone('Asia/Shanghai')),
         schedule_interval='00 11 * * *',
         concurrency=128,
         template_undefined=jinja2.Undefined,
         default_args={
             'owner': 'bigdata',
             # 'run_as_user': 'hive',
             'trigger_rule': TriggerRule.NONE_FAILED,
             'execution_timeout': timedelta(hours=5),
             'depends_on_past': False,
             'retries': 3,
             'retry_delay': timedelta(minutes=5),
             'max_retry_delay': timedelta(minutes=30),
             'retry_exponential_backoff': True,
             'email_on_failure': is_pro_env,
             'email_on_retry': False,
             # 'on_failure_callback'      : dingding_task_alert(dingding_conn_id='dingding_bigdata') if is_pro_env else None,
             'on_failure_callback': feishu_task_alert(feishu_conn_id='feishu_bigdata') if is_pro_env else None,
         },
         user_defined_macros=macros_dict,
         user_defined_filters=macros_dict,
         sla_miss_callback=combine([feishu_sla_alert(feishu_conn_id='feishu_bigdata')],
                                   ignore_error=True) if is_pro_env else None,
         on_success_callback=dingding_dag_alert(dingding_conn_id='dingding_bigdata'),
         on_failure_callback=feishu_dag_alert(feishu_conn_id='feishu_bigdata'),
         catchup=False) as dag:
    # import jms_data_back_up.ods
    import jms_data_back_up.dwd
    # import jms_data_back_up.dim
    import jms_data_back_up.time_sensor
    # import jms_data_back_up.dws
    # import jms_data_back_up.dm
    # import jms_data_back_up.datacheck
    import jms_data_back_up.doris
    # import jms_data_back_up.oss
    # import jms_data_back_up.es
    # import jms_data_back_up.hbase
    # import jms_data_back_up.oracle

    ods_tasks = [task for task in dag.tasks if
                 task.pool in ['oracle_tab', 'oracle_lmdm', 'oracle_tms', ] and len(task.upstream_list) == 0]
    if is_pro_env:
        for t in ods_tasks:
            t.sla = timedelta(hours=3)
            t.email_on_retry = True
            if t.on_retry_callback is None:
                t.on_retry_callback = combine([dingding_task_alert(dingding_conn_id='dingding_bigdata'),
                                               call_phone_by_email(voice_name='任务重跑')],
                                              ignore_error=True)
